package com.atguigu.gmall.realtime.test;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/5/1 19:58
 */
public class CoponUseDemo extends BaseSQLApp {
    
    public static void main(String[] args) {
        new CoponUseDemo().init("a", 10000, 1, "a");
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        tEnv.executeSql(
            "create table `topic_db`( " +
                "`database` string, " +
                "`table` string, " +
                "`data` map<string, string>, " +
                "`type` string, " +
                "`old` string," +
                "`ts` string " +
                ")" + SQLUtil.getKafkaDDL("dwd_tool_coupon_get", "ods_db"));
    
        // TODO 4. 读取优惠券领用数据，封装为表
        Table resultTable = tEnv.sqlQuery(
            "select " +
                "data['id'], " +
                "data['coupon_id'], " +
                "data['user_id'], " +
                "date_format(data['get_time'],'yyyy-MM-dd') date_id, " +
                "data['get_time'], " +
                "`old` " +
                "ts " +
                "from topic_db " +
                "where `table` = 'coupon_use' " +
                "and `type` = 'update' " +
                "and regexp(`old`, 'using_time') ");
        
        resultTable.execute().print();
        
    }
}
